package com.relytosoft.mizar.media.host.emqx;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.relytosoft.dubhe.common.core.constant.CommonConstants;
import com.relytosoft.dubhe.common.core.object.ResponseResult;
import com.relytosoft.mizar.media.common.constant.MediaMqConstants;
import com.relytosoft.mizar.media.common.emqx.client.MqttOp;
import com.relytosoft.mizar.media.common.entity.pojo.dto.FaceDispositionReqDto;
import com.relytosoft.mizar.media.common.entity.pojo.dto.TaskRequestDto;
import com.relytosoft.mizar.media.external.api.analyze.service.FaceService;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

/**
 * FaceMqttMessageListener
 *
 * @Author SunLi
 * @Date: 2022/5/31 10:22
 * @Version 1.0
 */
@Component
public class FaceMqttMessageListener implements com.relytosoft.mizar.media.common.emqx.listener.FaceMqttMessageListener {

    private static final Logger log = LoggerFactory.getLogger(FaceMqttMessageListener.class);

    @Resource
    public FaceService faceService;

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        try {
            boolean isConsume = false;
            String mqttMsg = new String(message.getPayload(), StandardCharsets.UTF_8);
            JSONObject jsonObject = JSON.parseObject(mqttMsg);
            String opType = jsonObject.getString("OpType");
            String data = jsonObject.getString("Data");
            //实时分析任务
            if (topic.equals(MediaMqConstants.FACE_TASK_TOPIC)) {
                switch (opType) {
                    //添加实时分析任务
                    case "insertTask":
                        isConsume = createTask(data);
                        break;
                    //删除实时分析任务
                    case "deleteTask":
                        isConsume = stopTask(data);
                        break;
                    default:
                        break;
                }
            } else if (topic.equals(MediaMqConstants.FACE_DISPOSITION_TOPIC)) {
                switch (opType) {
                    //添加布控任务
                    case "insertDisposition":
                        isConsume = createDisposition(data);
                        break;
                    //删除布控任务
                    case "deleteDisposition":
                        isConsume = deleteDisposition(data);
                        break;
                    default:
                        break;
                }
            }
            log.info("线程名称：【" + Thread.currentThread().getName() + "】" + "接收的主题：" + topic + ",接收的操作类型：" + opType + "，执行结果：" + isConsume);
        } catch (Exception e) {
            log.info("线程名称：【" + Thread.currentThread().getName() + "】" + "接收的主题：" + topic + ",处理发生错误：" + e.getMessage());
        }
    }

    /**
     * 创建实时分析任务
     * @param data 任务信息
     * @return 消费结果
     */
    private boolean createTask(String data){
        TaskRequestDto taskRequestDto=JSON.parseObject(data,TaskRequestDto.class);
        ResponseResult result=faceService.createTask(taskRequestDto);
        if(result.getCode()== CommonConstants.SUCCESS){
            //发送实时分析子任务ID同步消息
            JSONObject jsonObject=new JSONObject();
            jsonObject.put("OpType","updateTaskId");
            jsonObject.put("Data", result.getData());
            MqttOp.publish(MediaMqConstants.FACE_DATA_TOPIC,JSON.toJSONString(jsonObject));
            return true;
        }
        //创建任务调用接口失败，需要发送消息更新任务为异常，以待后续重新启动或删除
        Map<String, Object> info = new HashMap<>(2);
        info.put("id", taskRequestDto.getId());
        info.put("msg", result.getMsg());
        //发送消息
        JSONObject mqttMsg=new JSONObject();
        mqttMsg.put("OpType","taskUnusual");
        mqttMsg.put("Data", info);
        MqttOp.publish(MediaMqConstants.FACE_DATA_TOPIC,JSON.toJSONString(mqttMsg));
        log.info("人脸实时分析任务同步添加bitman时,发生错误：" + result.getMsg());
        return false;
    }

    /**
     * 删除实时分析任务
     * @param data 任务ID值
     * @return 消费结果
     */
    private boolean stopTask(String data){
        ResponseResult result=faceService.stopTask(data);
        if(result.getCode()== CommonConstants.SUCCESS){
            return true;
        }
        log.info("人脸实时分析任务同步删除bitman时,发生错误：" + result.getMsg());
        return false;
    }
    /**
     * 人脸布控任务添加
     * @param data 布控信息
     * @return 消费结果
     */
    private boolean createDisposition(String data){
        FaceDispositionReqDto dispositionReqDto=JSON.parseObject(data,FaceDispositionReqDto.class);
        ResponseResult result=faceService.createDisposition(dispositionReqDto);
        if(result.getCode()== CommonConstants.SUCCESS){
            //发送布控任务ID同步消息
            JSONObject jsonObject=new JSONObject();
            jsonObject.put("OpType","updateDispositionId");
            jsonObject.put("Data", result.getData());
            MqttOp.publish(MediaMqConstants.FACE_DATA_TOPIC,JSON.toJSONString(jsonObject));
            return true;
        }
        log.info("人脸布控任务同步添加bitman时,发生错误：" + result.getMsg());
        return false;
    }

    /**
     * 人脸布控任务删除
     * @param data 布控信息
     * @return 消费结果
     */
    private boolean deleteDisposition(String data){
        ResponseResult result=faceService.deleteDisposition(data);
        if(result.getCode()== CommonConstants.SUCCESS){
            return true;
        }
        log.info("人脸布控任务同步删除bitman时,发生错误：" + result.getMsg());
        return false;
    }
}
